package com.blue.rpc.client;

import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.blue.rpc.core.message.RpcRequest;
import com.blue.rpc.util.RpcConst;
import com.blue.rpc.util.RpcException;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyRpcClient implements RpcClient
{
	private static Logger logger = LoggerFactory.getLogger(NettyRpcClient.class);

	private final int reconnect;
	private final int timeout;
	private final ChannelInitializer<SocketChannel> initializer;

	private final Map<String, RpcClientHandler> handlerMap = new ConcurrentHashMap<>();
	private final Map<String, ServiceHost> serviceMap = new ConcurrentHashMap<>();

	private Bootstrap bootstrap;
	private EventLoopGroup workerGroup;
	
	public NettyRpcClient(String token)
	{
		this(RpcConst.RECONNECT, RpcConst.TIMEOUNT, token);
	}

	public NettyRpcClient(int reconnect, int timeout, String token)
	{
		this.reconnect = reconnect;
		this.timeout = timeout;
		this.initializer = new RpcClientInitializer(this, token);
	}

	@Override
	public void start()
	{
		workerGroup = new NioEventLoopGroup(4);
		bootstrap = new Bootstrap();
		bootstrap.group(workerGroup).channel(NioSocketChannel.class)
				.option(ChannelOption.SO_KEEPALIVE, true)
				.option(ChannelOption.TCP_NODELAY, true)
				.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeout)
				.handler(initializer);
	}

	public void updateConnect(Map<String, List<String>> serviceMap)
	{
		Set<String> hostSet = new HashSet<>();
		handlerMap.forEach((k, v) -> hostSet.add(k));

		for (Map.Entry<String, List<String>> entry : serviceMap.entrySet())
		{
			String service = entry.getKey();
			for (String host : entry.getValue())
			{
				if (handlerMap.containsKey(host)) // 已经有连接
				{
					hostSet.remove(host);
				}
				else
				{
					this.connect(service, host);
				}
			}
		}

		for (String host : hostSet) // 关闭连接
		{
			RpcClientHandler handler = handlerMap.get(host);
			if (handler == null)
				continue;

			handler.close();
			for (Map.Entry<String, ServiceHost> entry : this.serviceMap.entrySet())
			{
				entry.getValue().removeHost(host);
			}
		}
	}

	public void connect(final String service, final String host)
	{
		String[] hosts = host.split(":");
		if (hosts.length != 2)
		{
			logger.warn("远程地址错误：{}", host);
			return;
		}
		InetSocketAddress remote = new InetSocketAddress(hosts[0],
				Integer.parseInt(hosts[1]));
		bootstrap.connect(remote).addListener(new ChannelFutureListener()
		{
			@Override
			public void operationComplete(ChannelFuture future) throws Exception
			{
				if (future.isSuccess())
				{
					synchronized (NettyRpcClient.this)
					{
						ServiceHost serviceHost = serviceMap.get(service);
						if (serviceHost == null)
						{
							serviceHost = new ServiceHost(service);
							serviceMap.put(service, serviceHost);
						}
						serviceHost.addHost(host);
					}

					logger.info("成功连接远程服务器：{}", remote);
					return;
				}

				logger.warn("连接远程服务器失败：{}，{} ms后重新连接", remote, reconnect);
				future.channel().eventLoop().schedule(() -> start(), reconnect,
						TimeUnit.MILLISECONDS);

			}
		});
	}

	@Override
	public void stop()
	{
		for (Map.Entry<String, RpcClientHandler> entry : handlerMap.entrySet())
		{
			entry.getValue().close();
		}
		if (workerGroup != null)
		{
			workerGroup.shutdownGracefully();
		}
	}

	@Override
	public RpcClientFuture invoke(RpcRequest request) throws RpcException
	{
		RpcClientHandler handler = this.getHandler(request);
		return handler.invoke(request);
	}

	@Override
	public RpcClientFuture invoke(RpcRequest request, RpcClientCallback callback)
			throws RpcException
	{
		RpcClientHandler handler = this.getHandler(request);
		return handler.invoke(request, callback);
	}

	private RpcClientHandler getHandler(RpcRequest request)
	{
		if (handlerMap.isEmpty() || serviceMap.isEmpty())
			throw new RpcException("找不到可用的服务列表");

		ServiceHost serviceHost = serviceMap.get(request.getService());
		if (serviceHost == null)
			throw new RpcException("找不到可用的服务：" + request.getService());

		String host = serviceHost.getHost();
		if (host == null)
			throw new RpcException("找不到服务的远程主机：" + request.getService());

		RpcClientHandler handler = handlerMap.get(host);
		if (handler == null)
			throw new RpcException("找不到服务的处理器：" + request.getService());

		return handler;
	}

	public void put(String remote, RpcClientHandler handler)
	{
		handlerMap.put(remote, handler);
		logger.info("添加远程服务器：{}－handler 个数：{}", remote, handlerMap.size());
	}

	public void remove(String remote)
	{
		handlerMap.remove(remote);
		logger.info("移除远程服务器：{}－handler 个数：{}", remote, handlerMap.size());
	}

}
